package org.apache.hadoop.hdfs.server.datanode;

import java.io.IOException;
import java.net.SocketTimeoutException;
import java.nio.channels.AsynchronousCloseException;
import java.util.HashMap;

import org.apache.commons.logging.Log;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.net.Peer;
import org.apache.hadoop.hdfs.net.PeerServer;
import org.apache.hadoop.hdfs.util.DataTransferThrottler;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Daemon;

import com.google.common.annotations.VisibleForTesting;

/**
 * DataXceiverServer是数据节点DataNode上一个用于接收数据读写请求的后台工作线程，为每个数据读写请求(客户端或datanode)创建一个单独的线程去处理。
 * 它提供了 一请求一线程 的模式，并对线程数目做了控制，对接收数据读写请求时发生的各种异常做了很好的容错处理，
 * 特别是针对内存溢出异常，允许等待短暂时间再继续提供服务，避免内存使用高峰期等等。
 * 而且，线程组与后台线程的应用，也大大简化可这些线程的管理工作；
 * 对数据读写请求处理线程的数目，集群内数据块移动线程数目都做了严格控制，避免资源无节制耗费等。
 * 这些设计很好的支撑了HDFS大吞吐量数据的性能要求，可以说，是一个很好的设计方案，值得我们在其他类似需求的系统中借鉴。
 *
 * DataXceiverServer是一个线程，专门负责接收或者发送一个数据block,
 * 这个东西是创建了以后专门用来负责监听从hdfs客户端或者其他datanode发送过来的请求
 * 这个东西是走底层socket网络连接和请求的，不是走hadoop的rpc远程接口调用的
 */
class DataXceiverServer implements Runnable {
  public static final Log LOG = DataNode.LOG;

  /**
   * PeerServer类型的peerServer，是DataXceiverServer实现功能最重要的一个类，
   * 在DataXceiverServer实例构造时，实际上传入的是实现了PeerServer接口的TcpPeerServer类，
   * 该类内部封装了一个ServerSocket，提供了Java Socket服务端的功能，用于监听来自客户端或其他DataNode的数据读写请求。
   */
  private final PeerServer peerServer;
  /**
   * 该DataXceiverServer所属DataNode实例datanode ,
   * 该线程就能随时获得DataNode状态、提供的一些列服务等
   */
  private final DataNode datanode;
  // Peer 实际上就是对Socket的封装；
  /** Peer所在线程的映射集合peers */
  private final HashMap<Peer, Thread> peers = new HashMap<Peer, Thread>();
  /** Peer与DataXceiver的映射集合peersXceiver */
  private final HashMap<Peer, DataXceiver> peersXceiver = new HashMap<Peer, DataXceiver>();
  /** DataXceiverServer是否已关闭的标志位closed */
  private boolean closed = false;
  
  /**
   * 每个节点并行的最大DataXceivers数目。
   * 为了避免dataNode运行内存溢出，执行这个限制是必须的。
   * 定义是默认值为4096.
   */
  int maxXceiverCount = DFSConfigKeys.DFS_DATANODE_MAX_RECEIVER_THREADS_DEFAULT;

  /**
   * 确保集群平衡不占用太多资源的一种手段或管理者。
   * 它限制了为集群平衡所做的块移动的数量及它们所占用的总宽带，是一种节流器的概念。
   *
   * 它内部有三个非常重要的指标，
   * 1. 表示当前移动数据块的线程数的 numThreads，
   * 2. 表示移动数据块的最大线程数 maxThreads，
   * 3. 数据传输的带宽 bandwidth。
   *
   * 在其构造方法内，会调用父类的构造方法设置带宽bandwidth，并在子类中设置移动数据块的最大线程数maxThreads。
   */
  static class BlockBalanceThrottler extends DataTransferThrottler {
   private int numThreads; // 表示当前移动数据块的线程数 numThreads
   private int maxThreads; // 表示移动数据块的最大线程数 maxThreads
   
   /**
    * 构造方法
    * @param bandwidth 可用于平衡的总带宽量
    */
   private BlockBalanceThrottler(long bandwidth, int maxThreads) {
     super(bandwidth);
     this.maxThreads = maxThreads; // 设置移动数据块的最大线程数maxThreads
     LOG.info("Balancing bandwith is "+ bandwidth + " bytes/s");
     LOG.info("Number threads for balancing is "+ maxThreads);
   }
   
   /**
    * 检测block移动是否可以开始
    * 在移动数据块block之前，会调用acquire()方法，确认一个数据块是否可以移动。
    *     实际上是当前线程数numThreads大于等于最大线程数maxThreads时，返回false，block不可以移动；
    *     否则，当前线程数numThreads累加，并返回true，block可以移动。
    */
   synchronized boolean acquire() {
     // 当前线程数numThreads大于等于最大线程数maxThreads时，返回false，block不可以移动
     if (numThreads >= maxThreads) {
       return false;
     }
     // 否则，当前线程数numThreads累加，并返回true，block可以移动
     numThreads++;
     return true;
   }
   
   /**
    * 当数据块移动完毕后，则调用release()方法，标志移动已完成，线程计数器减一
    */
   synchronized void release() {
     // 当前线程数numThreads减1
     numThreads--;
   }
  }

  /** 集群数据块平衡节流器balanceThrottler */
  final BlockBalanceThrottler balanceThrottler;
  
  /**
   * 我们需要估计块大小以检测磁盘分区是否有足够的空间。
   * 新客户端传递预期块大小给DataNode。
   * 对于旧客户端而言我们仅仅使用服务器端默认的块大小。
   */
  final long estimateBlockSize;


  /**
   * 在构造DataXceiverServer时，会根据传入的peerServer设置同名成员变量、设置DataNode实例datanode等，并初始化两个重要的指标，
   * 第一个是设置DataNode中DataXceiver的最大数目maxXceiverCount，它取参数dfs.datanode.max.transfer.threads，参数未配置的话，默认值为4096；
   * 第二个便是设置估计块大小estimateBlockSize，它取参数取参数dfs.blocksize，参数未配置的话，默认值是128*1024*1024，即128M，
   * 最后，设置集群平衡节流器，带宽取参数dfs.datanode.balance.bandwidthPerSec，参数未配置默认为1024*1024，
   * 最大线程数取参数dfs.datanode.balance.max.concurrent.moves，参数未配置默认为5。
   */
  DataXceiverServer(PeerServer peerServer, Configuration conf, DataNode datanode) {
    // 根据传入的peerServer设置同名成员变量
    this.peerServer = peerServer;
    // 设置DataNode实例datanode
    this.datanode = datanode;

    // 设置DataNode中DataXceiver的最大数目maxXceiverCount
    // 取参数dfs.datanode.max.transfer.threads，参数未配置的话，默认值为4096
    this.maxXceiverCount =  conf.getInt(DFSConfigKeys.DFS_DATANODE_MAX_RECEIVER_THREADS_KEY, DFSConfigKeys.DFS_DATANODE_MAX_RECEIVER_THREADS_DEFAULT);

    // 设置估计块大小estimateBlockSize
    // 取参数dfs.blocksize，参数未配置的话，默认值是128*1024*1024，即128M
    this.estimateBlockSize = conf.getLongBytes(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, DFSConfigKeys.DFS_BLOCK_SIZE_DEFAULT);

    // 设置集群平衡节流器
    // 带宽取参数dfs.datanode.balance.bandwidthPerSec，参数未配置默认为 1024*1024
    // 最大线程数取参数dfs.datanode.balance.max.concurrent.moves，参数未配置默认为5
    this.balanceThrottler = new BlockBalanceThrottler( //
        conf.getLong(DFSConfigKeys.DFS_DATANODE_BALANCE_BANDWIDTHPERSEC_KEY, DFSConfigKeys.DFS_DATANODE_BALANCE_BANDWIDTHPERSEC_DEFAULT),
        conf.getInt(DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY, DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT));
  }


  /**
   * 核心方法:
   * 当datanode正常运转的时候，DataXceiverServer线程主要负责在一个while循环中利用TcpPeerServer（也就是ServerSocket）的accept()方法阻塞，
   * 直到接收到客户端或者其他DataNode的连接请求，然后：
   *      1、获得peer，即Socket的封装；
   *      2、判断当前DataNode上DataXceiver线程数量是否超过阈值，如果超过的话，直接抛出IOException，利用IOUtils的cleanup()方法关闭peer后继续循环，否则继续3；
   *      3、创建一个后台线程DataXceiver，并将其加入到datanode的线程组threadGroup中，并启动该线程，响应数据读写请求；
   */
  @Override
  public void run() {
    Peer peer = null;
    // 如果标志位shouldRun为true，且没有为升级而执行shutdown
    while (datanode.shouldRun && !datanode.shutdownForUpgrade) {
      try {
        // 阻塞，直到接收到客户端或者其他DataNode的连接请求
        peer = peerServer.accept(); // 线程一旦启动，就会while true循环，每次循环到这里，peerServer.accept();会等到别人建立socket连接，

        // 确保DataXceiver数目没有超过最大限制
        /**
         * DataNode的getXceiverCount方法计算得到，返回线程组的活跃线程数目
         * threadGroup == null ? 0 : threadGroup.activeCount();
         */
        int curXceiverCount = datanode.getXceiverCount();
        if (curXceiverCount > maxXceiverCount) {
          throw new IOException("Xceiver count " + curXceiverCount + " exceeds the limit of concurrent xcievers: " + maxXceiverCount);
        }

        // 只要有人跟它建立了一个连接，创建一个后台线程，专门负责这个连接的所有请求和响应
        // 如果有个hdfs客户端来连接这台datanode，那么它收到一个连接请求之后，就会创建爱你一个DataXceiver线程来处理跟这个hdfs客户端的所有请求和响应
        // DataXceiver，并加入到线程组datanode.threadGroup
        new Daemon(datanode.threadGroup, DataXceiver.create(peer, datanode, this)).start();
      } catch (SocketTimeoutException ignored) { //  1、SocketTimeoutException：Socket连接超时异常，忽略直接进入下一个循环即可，继续阻塞侦听请求；
        // 等待唤醒看看是否能够继续运行
      } catch (AsynchronousCloseException ace) { // 2、AsynchronousCloseException：异步关闭异常，这里我们需要通过判断DataNode的状态来决定是否继续进行循环，继续阻塞侦听请求；
        // 正如我们所预料的，只有在关机的过程中，通过其他线程关闭我们的侦听套接字，其他情况下则不会发生

      } catch (IOException ie) { // 3、IOException：此时，需要关闭Socket后进入下一个循环，继续阻塞侦听请求；
        IOUtils.cleanup(null, peer);
        LOG.warn(datanode.getDisplayName() + ":DataXceiverServer: ", ie);

      } catch (OutOfMemoryError ie) {
        // 4、OutOfMemoryError：内存溢出错误，这种情况下数据节点可能由于存在太多的数据传输导致内存溢出，记录该事件，并等待30秒，其他的数据传输可能到时就完成了。
        // 我们需要做的就是，首先需要利用IOUtils的cleanup()方法关闭peer，记录警告日志信息，
        // 然后线程休眠30秒，等待DataNode其他数据读写服务完成后，进入下一个循环， 继续阻塞侦听请求；

        IOUtils.cleanup(null, peer);
        // 数据节点可能由于存在太多的数据传输导致内存溢出，记录该事件，并等待30秒，其他的数据传输可能到时就完成了
        LOG.warn("DataNode is out of memory. Will retry in 30 seconds.", ie);
        try {
          Thread.sleep(30 * 1000);
        } catch (InterruptedException e) {
        }

      } catch (Throwable te) {
        // 5、Throwable：DataNode就是为提高数据存储、读写服务而生的，既然出现了Throwable，那么DataNode也应该停止了，记录error日志信息，设置datanode的shouldRun为false，退出循环。
        LOG.error(datanode.getDisplayName() + ":DataXceiverServer: Exiting due to: ", te);
        datanode.shouldRun = false;
      }
    }

    // 当Throwable发生，退出循环后，我们就需要做一些列的关闭操作，

    // 关闭服务器停止接收更多请求
    try {
      peerServer.close(); // 关闭peerServer，
      closed = true; // 设置DataXceiverServer标志位closed为true
    } catch (IOException ie) {
      LOG.warn(datanode.getDisplayName() + " :DataXceiverServer: close exception", ie);
    }

    // 如果在重新启动前准备阶段，在关闭前通知peers
    if (datanode.shutdownForUpgrade) {
      // 如果在重新启动前准备阶段，在关闭peers之前，需要先通知它们，
      // 通知的方式就是通过调用restartNotifyPeers()方法，获取peers的每个peer所在线程，通过interrupt()方法打断它们
      restartNotifyPeers(); //
      LOG.info("Shutting down DataXceiverServer before restart");
      // 每个线程需要一些时间去完成自己。如果一个线程需要发送OOB至客户端，但是在网络上被阻塞的了一段时间，我们需要强迫使其停止。此时，我们需要大约2秒的时间等待它们的完成。
      for (int i = 0; getNumPeers() > 0 && i < 10; i++) {
        try {
          Thread.sleep(200);
        } catch (InterruptedException e) {
        }
      }
    }

    // 关闭所有的peers
    closeAllPeers(); //
  }

  /**
   * 杀死DataXceiverServer线程
   */
  void kill() {
    // DataXceiverServer线程被kill时，需要确定datanode的标志位shouldRun为false，或者标志位shutdownForUpgradetrue
    // 也就意味着，当datanode不应该继续运行，或者为了升级而关闭时，DataXceiverServer线程才可以被kill
    assert (datanode.shouldRun == false || datanode.shutdownForUpgrade) : "shoudRun should be set to false or restarting should be true before killing";
    try {
      // 关闭peerServer，即关闭ServerSocket
      this.peerServer.close();
      // 设置标志位closed为true
      this.closed = true;
    } catch (IOException ie) {
      LOG.warn(datanode.getDisplayName() + ":DataXceiverServer.kill(): ", ie);
    }
  }

  /**
   * 添加一个Peer
   */
  synchronized void addPeer(Peer peer, Thread t, DataXceiver xceiver) throws IOException {
    // 首先判断DataXceiverServer线程的标志位closed，为true时，说明服务线程已被关闭，不能再提供Socket通讯服务
    if (closed) {
      throw new IOException("Server closed.");
    }
    // 将peer与其所在线程t的映射关系加入到peers中
    peers.put(peer, t);
    // 将peer与其所属DataXceiver xceiver映射关系加入到peersXceiver中
    peersXceiver.put(peer, xceiver);
  }

  /**
   * 关闭一个Peer
   */
  synchronized void closePeer(Peer peer) {
    peers.remove(peer);
    // 从数据结构peers、peersXceiver移除peer对应记录
    peersXceiver.remove(peer);
    // 利用IOUtils的cleanup关闭peer，即关闭socket
    IOUtils.cleanup(null, peer);
  }

  // Sending OOB to all peers
  public synchronized void sendOOBToPeers() {
    if (!datanode.shutdownForUpgrade) {
      return;
    }

    for (Peer p : peers.keySet()) {
      try {
        peersXceiver.get(p).sendOOB();
      } catch (IOException e) {
        LOG.warn("Got error when sending OOB message.", e);
      } catch (InterruptedException e) {
        LOG.warn("Interrupted when sending OOB message.");
      }
    }
  }
  
  // 通知所有peers关闭并重新启动。
  // datanode.shouldRun 应该还是 true , datanode.restarting 应在调用此方法之前设置为true。
  synchronized void restartNotifyPeers() {
    assert (datanode.shouldRun == true && datanode.shutdownForUpgrade);
    for (Peer p : peers.keySet()) {
      // 中断每个DataXceiver线程
      peers.get(p).interrupt();
    }
  }

  /**
   * 关闭所有Peer
   */
  synchronized void closeAllPeers() {
    // 记录info日志信息
    LOG.info("Closing all peers.");
    // 循环关闭所有的peer
    for (Peer p : peers.keySet()) {
      IOUtils.cleanup(LOG, p);
    }
    // 清空peer数据集合
    peers.clear();
    peersXceiver.clear();
  }

  // Return the number of peers.
  synchronized int getNumPeers() {
    return peers.size();
  }

  // Return the number of peers and DataXceivers.
  @VisibleForTesting
  synchronized int getNumPeersXceiver() {
    return peersXceiver.size();
  }
  
  synchronized void releasePeer(Peer peer) {
    peers.remove(peer);
    peersXceiver.remove(peer);
  }
}
